package DianShang_2024.ds_06.extract


import org.apache.hudi.DataSourceWriteOptions.{PARTITIONPATH_FIELD, PRECOMBINE_FIELD, RECORDKEY_FIELD}
import org.apache.hudi.QuickstartUtils.getQuickstartWriteConfigs
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, date_format, greatest, lit, max, when}

import java.util.Properties

object extract01 {
  def main(args: Array[String]): Unit = {
          /*
                抽取shtd_store库中user_info的增量数据进入Hudi的ods_ds_hudi库中表user_info。根据ods_ds_hudi.user_info表
                中operate_time或create_time作为增量字段(即MySQL中每条数据取这两个时间中较大的那个时间作为增量字段去和ods_ds_hudi里
                的这两个字段中较大的时间进行比较)，只将新增的数据抽入，字段名称、类型不变，同时添加分区，若operate_time为空，则
                用create_time填充，分区字段为etl_date，类型为String，且值为当前比赛日的前一天日期（分区字段格式为yyyyMMdd）。id作
                为primaryKey，operate_time作为preCombineField。使用spark-shell执行show partitions ods_ds_hudi.user_inf
                o命令，将结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下
           */

    //  准备spark集成hudi的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据抽取第一题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use ods_ds_hudi")

    //  设置连接mysql的配置
    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  hdfs里面保存hudi表的路径
 val user_info_path="hdfs://192.168.40.110:9000/user/hive/warehouse/ods_ds_hudi.db/user_info"

 //  将ods_ds_hudi里面的user_info的数据读取到，方便后面使用sql将最大的时间的值拿出来
 spark.read.format("hudi").load(user_info_path)
   .createOrReplaceTempView("user_info")

//  collect()将表格收集为一行一行的数据,所以collect()(0)为第一行的数据,然后使用get(0)为第一列的数据
    val max_time=spark.sql(
      """
        |select
        |if(create_time > operate_time,create_time,operate_time) as max_time
        |from user_info
        |order by max_time desc
        |""".stripMargin).collect()(0).get(0).toString
    println("sql语句拿到的最大的时间值:",max_time)

    //  使用dataframe的形式拿到最大的时间值
    //  这个地方要注意的是下面的max,greatest,col导包必须是spark下面的，否则agg函数会报错
    //   这里需要尤其注意的是max函数，必须要导入org.apache.spark.sql.functions.max，否则agg函数会报错，因为快捷键默认是导入错误的包
    //  导入的是breeze.linalg.max
    /*
        1.agg(...): 这是一个 DataFrame 的方法，用于执行聚合操作。它接收一个或多个聚合函数作为参数，并返回一个新的 DataFrame，其中包含聚合结果。
        2.max(...): 这是一个聚合函数，用于计算指定列的最大值。
       3.greatest(col("create_time"),col("operate_time")): 这是一个表达式，它使用 greatest 函数来选择 create_time 和 operate_time
        列中的较大值,并且返回较大的，然后形成的一列数据再用max函数求出这一列最大的就是整个表最大时间。col 函数用于引用 DataFrame 中的列。
        first:是dataframe的函数，返回第一行数据
         get(0):也是dataframe的函数，用于返回第一列的数据
         greatest:最大的
     */
    val MaxTime=spark.read.format("hudi").load(user_info_path)
      .agg(max(greatest(col("create_time"),col("operate_time"))))
      .first()
      .get(0)
      .toString

    println("Dataframe函数得到的最大时间:",MaxTime)

    //  根据条件抽取数据
    /*
        withColumn:新增不存在的列，替换已经存在的列，下面的用法是替换
        when(col("operate_time").isNull,col("create_time")):如果operate_time为空，则返回create_time的值
        otherwise(col("operate_time")):如果前面when的条件不满足，则返回otherwise里面的值
        greatest 是一个 SQL 函数，用于返回其参数中的最大值。在这里，它比较 "create_time" 和 "operate_time" 两个列的值
        where:进行过滤，保留符合条件的数据，下面保留的是时间的最大值，大于MaxTime的数据
        这里根据条件筛选出数据之后，也需要将时间类型的数据使用date_format()函数转换一下再写进去，否则会变成时间戳，而且时间不对
        greatest:最大的

       之所以在where函数做比较的对象需要用lit函数，是因为spark的dataframe在使用where函数进行比较时，需要确保比较的对象是和dataframe
       有同样的列的兼容的数据类型，所以需要使用lit
     */
    spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","user_info",connect)
      .withColumn(
        "operate_time",
        when(col("operate_time").isNull,col("create_time")).otherwise(col("operate_time"))
      )
      .where(
        greatest(col("create_time"),col("operate_time"))  >   lit(MaxTime).cast("timestamp")
      )
      .withColumn("etl_date",lit("20240101"))
      .withColumn("create_time",date_format(col("create_time"),"yyyy-MM-dd HH:mm:ss"))
      .withColumn("operate_time",date_format(col("create_time"),"yyyy-MM-dd HH:mm:ss"))
      .write.mode("append").format("hudi")
      .options(getQuickstartWriteConfigs)
      .option(PRECOMBINE_FIELD.key(),"operate_time")
      .option(RECORDKEY_FIELD.key(),"id")
      .option(PARTITIONPATH_FIELD.key(),"etl_date")
      .option("hoodie.table.name","user_info")
      .save(user_info_path)





//   直接使用sql的方式将数据插入到hudi表是不行的，会报错，因为hudi表自带的字段，hive和mysql表没有

//      //  读取到mysql里面的数据读取到并创建临时表mysql_user_info
//    spark.read.jdbc("jdbc:mysql://192.168.40.110/shtd_store?useSSL=false","user_info",connect)
//      .createOrReplaceTempView("mysql_user_info")
//
//    //    将数据抽取到hudi表
//    spark.sql(
//      s"""
//        |insert into table user_info
//        |partition(etl_date='20231027')
//        |select
//        |*
//        |from mysql_user_info
//        |where if(create_time > operate_time,create_time,operate_time) >  cast('${max_time}' as timestamp)
//        |""".stripMargin).show









    //   关闭环境
    spark.close()
  }

}
